package com.winning.devops.batch.reader.ch07;

import com.winning.devops.batch.reader.ch07.domain.Customer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.batch.item.*;
import org.springframework.batch.item.file.FlatFileItemReader;
import org.springframework.batch.item.file.mapping.DefaultLineMapper;
import org.springframework.batch.item.file.mapping.FieldSetMapper;
import org.springframework.batch.item.file.transform.DelimitedLineTokenizer;
import org.springframework.batch.item.file.transform.FieldSet;
import org.springframework.core.io.ClassPathResource;
import org.springframework.stereotype.Service;
import org.springframework.validation.BindException;

/**
 * 重启读取
 *
 * @author chensj
 * @date 2020-03-04 21:28
 */
@Service("restartReader")
public class RestartReader implements ItemStreamReader<Customer> {

    /**
     * logger
     */
    private static final Logger logger = LoggerFactory.getLogger(RestartReader.class);

    private FlatFileItemReader<Customer> customerFlatFileItemReader = new FlatFileItemReader<>();
    private Integer currentLine = 0;
    private boolean restart = false;
    private ExecutionContext executionContext;

    public RestartReader() {
        // 初始化Reader
        // 设置读取文件位置
        customerFlatFileItemReader.setResource(new ClassPathResource("restart.txt"));
        // 跳过第几行
        customerFlatFileItemReader.setLinesToSkip(1);
        // 数据解析
        DelimitedLineTokenizer tokenizer = new DelimitedLineTokenizer();
        tokenizer.setNames(new String[]{"id", "firstName", "lastName", "birthday"});
        // 映射到实体类
        DefaultLineMapper<Customer> mapper = new DefaultLineMapper<>();
        mapper.setLineTokenizer(tokenizer);
        mapper.setFieldSetMapper(new FieldSetMapper<Customer>() {
            @Override
            public Customer mapFieldSet(FieldSet fieldSet) throws BindException {
                Customer customer = new Customer();
                customer.setId(fieldSet.readInt("id"));
                customer.setFirstName(fieldSet.readString("firstName"));
                customer.setLastName(fieldSet.readString("lastName"));
                customer.setBirthday(fieldSet.readString("birthday"));
                return customer;
            }
        });
        // 检查参数
        mapper.afterPropertiesSet();
        customerFlatFileItemReader.setLineMapper(mapper);
    }

    @Override
    public Customer read() throws Exception, UnexpectedInputException, ParseException, NonTransientResourceException {
        Customer customer = null;
        this.currentLine++;
        if (restart) {
            // 跳过错误行
            customerFlatFileItemReader.setLinesToSkip(this.currentLine - 1);
            restart = false;
            logger.error("Start reading from line: {} ", this.currentLine);
        }
        customerFlatFileItemReader.open(this.executionContext);
        customer = customerFlatFileItemReader.read();
        // 手动抛出异常
        if (customer != null && customer.getFirstName().equals("c20")) {
            throw new RuntimeException("Error, Something wrong. Customer Id: " + customer.getId());
        }
        return customer;
    }

    /**
     * step执行之前执行
     *
     * @param executionContext 上下文
     * @throws ItemStreamException 异常
     */
    @Override
    public void open(ExecutionContext executionContext) throws ItemStreamException {
        this.executionContext = executionContext;
        // 判断是否存在参数 currentLine
        if (executionContext.containsKey("currentLine")) {
            this.currentLine = executionContext.getInt("currentLine");
            restart = true;
        } else {
            this.currentLine = 0;
            executionContext.put("currentLine", this.currentLine);
            System.out.println("Start reading from line: " + (this.currentLine + 1));
        }
    }

    /**
     * 一次chunk 读取完成后会执行这个代码
     * 注意 第一次读取成功后第二次失败，就不会执行这个代码
     *
     * @param executionContext 上下文
     * @throws ItemStreamException 异常
     */
    @Override
    public void update(ExecutionContext executionContext) throws ItemStreamException {
        // 数据会被存储在数据库中，下次会从这个数开始继续向下读取
        executionContext.put("currentLine", this.currentLine);
        System.out.println("update current line: " + (executionContext.getInt("currentLine")));
    }

    /**
     * 整个step 执行完成后执行
     *
     * @throws ItemStreamException 异常
     */
    @Override
    public void close() throws ItemStreamException {
        System.out.println("close current line" + (executionContext.getInt("currentLine")));
    }
}
